package com.qianjun.hadoop.test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.zookeeper.common.IOUtils;

import com.qianjun.hadoop.test.PullFilesFilter.RegexPathFilter;

public class MergeSmallFiles {

	private static FileSystem fs;
	private static FileSystem local;

	public static void main(String[] args) throws URISyntaxException, IOException {
		// TODO Auto-generated method stub

		list();
		
	}

	public static void list() throws URISyntaxException, IOException {
		Configuration conf = new Configuration();

		URI uri = new URI("hdfs://qianjun01:9000/");
		fs = FileSystem.get(uri, conf);
		local = FileSystem.getLocal(conf);

		FileStatus[] status = local.globStatus(new Path("C://Users/qianjun/Desktop/73/*"), new RegexExcludePathFilter("^.*svn$"));
		
		Path[] dirs = FileUtil.stat2Paths(status);
		
		FSDataOutputStream out =null;
		FSDataInputStream in =null;
		for(Path dir:dirs){
			String filename = dir.getName().replaceAll("-", "");
			
			FileStatus[] localStatus = local.globStatus(new Path(dir+"/*"), new RegexAcceptPathFilter("^.*txt$"));
			
			Path[] listFiles = FileUtil.stat2Paths(localStatus);
			Path block = new Path("hdfs://qianjun01:9000/qianjun/smallfile/"+filename+".txt");
			out = fs.create(block);
			
			for(Path file:listFiles){
				in = local.open(file);
				IOUtils.copyBytes(in, out, 4096, false);
				in.close();
			}
			
			if(out!=null){
				out.close();
			}
			
		}
	
	}

	public static class RegexExcludePathFilter implements PathFilter {

		private String regex = "";

		public RegexExcludePathFilter(String regex) {

			this.regex = regex;
		}

		@Override
		public boolean accept(Path path) {
			// TODO Auto-generated method stub

			boolean flag = path.toString().matches(regex);

			return !flag;
		}

	}
	
	public static class RegexAcceptPathFilter implements PathFilter {

		private String regex = "";

		public RegexAcceptPathFilter(String regex) {

			this.regex = regex;
		}

		@Override
		public boolean accept(Path path) {
			// TODO Auto-generated method stub

			boolean flag = path.toString().matches(regex);

			return flag;
		}

	}

}
